package com.thb.pay.mq;

import com.alibaba.fastjson.JSON;
import com.thb.pay.mq.consumer.MessageConsumer;
import com.thb.pay.mq.process.PoolExecutorProcess;
import com.thb.pay.mq.sender.MessageSender;
import com.thb.pay.mq.utils.MessageWithTime;
import com.thb.pay.mq.enums.ExchangeTypeEnum;
import com.thb.pay.mq.process.MessageProcess;
import com.thb.pay.mq.sender.ExecutorMessageSender;
import com.thb.pay.mq.utils.DetailRes;
import com.thb.pay.mq.utils.MqConstants;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.core.env.Environment;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * Created by littlersmall on 16/5/11.
 */
@Slf4j
public class MQAccessBuilder {
    final MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
    final MessageConverter messageConverter = new Jackson2JsonMessageConverter();
    private ConnectionFactory connectionFactory;

    private Environment environment;

    public MQAccessBuilder(ConnectionFactory connectionFactory, Environment environment) {
        this.connectionFactory = connectionFactory;
        this.environment = environment;
    }

    public MessageSender buildFountMessageSender(final String exchange, final String routingKey, final String queue) throws IOException {
        return buildMessageSender(exchange, routingKey, queue, ExchangeTypeEnum.FANOUT);
    }

    public MessageSender buildDirectMessageSender(final String exchange, final String routingKey, final String queue) throws IOException {
        return buildMessageSender(exchange, routingKey, queue, ExchangeTypeEnum.DIRECT);
    }

    public MessageSender buildTopicMessageSender(final String exchange, final String routingKey) throws IOException {
        return buildMessageSender(exchange, routingKey, null, ExchangeTypeEnum.TOPIC);
    }

    public MessageSender buildFanoutMessageSender(final String exchange, final String routingKey) throws IOException {
        return buildMessageSender(exchange, routingKey, null, ExchangeTypeEnum.FANOUT);
    }

    //1 构造template, exchange, routingkey等
    //2 设置message序列化方法
    //3 设置发送确认
    //4 构造sender方法
    public MessageSender buildMessageSender(final String exchange, final String routingKey,
                                            final String queue, final ExchangeTypeEnum type) throws IOException {
        //1
        buildSendQueue(exchange, routingKey, queue, type, false);

        //4
        return new MessageSender() {
            Connection connection;
            Channel channel;
            ExecutorMessageSender sendMessageTask;

            {
                connection = connectionFactory.createConnection();
                channel = connection.createChannel(false);
                sendMessageTask = new ExecutorMessageSender(connection, channel, exchange, routingKey);
            }

            public DetailRes send(Object message) {
                sendMessageTask.addQueueMsg(message);
                return new DetailRes(true, "");
            }


            public DetailRes send(MessageWithTime messageWithTime) {
                try {
                    channel.basicPublish(exchange, routingKey, null, JSON.toJSONBytes(messageWithTime.getMessage()));
                } catch (Exception e) {
                    e.printStackTrace();
                    return new DetailRes(false, "");
                }

                return new DetailRes(true, "");
            }


        };
    }

    public <T> MessageConsumer buildMessageConsumer(String exchange, String routingKey, final String queue,
                                                    final MessageProcess<T> messageProcess, Class<T> clazz, boolean autoDelete) throws IOException {
        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, ExchangeTypeEnum.DIRECT, clazz, autoDelete);
    }

    public <T> MessageConsumer buildTopicMessageConsumer(String exchange, String routingKey, final String queue,
                                                         final MessageProcess<T> messageProcess, Class<T> clazz, boolean autoDelete) throws IOException {
        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, ExchangeTypeEnum.TOPIC, clazz, autoDelete);
    }

    public <T> MessageConsumer buildFanoutMessageConsumer(String exchange, String routingKey, final String queue,
                                                          final MessageProcess<T> messageProcess, Class<T> clazz, boolean autoDelete) throws IOException {
        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, ExchangeTypeEnum.FANOUT, clazz, autoDelete);
    }

    //1 创建连接和channel
    //2 设置message序列化方法
    //3 consume
    public <T> MessageConsumer buildMessageConsumer(String exchange, String routingKey, final String queue,
                                                    final MessageProcess<T> messageProcess, ExchangeTypeEnum type, Class<T> clazz, boolean autoDelete) throws IOException {

        final Connection connection = connectionFactory.createConnection();

        buildConsumerQueue(exchange, routingKey, queue, type, autoDelete);
        // 3
        return new MessageConsumer() {
            Channel channel;
            PoolExecutorProcess<T> poolExecutorProcess;

            {
                poolExecutorProcess = new PoolExecutorProcess<>(environment, messageProcess, queue);
                channel = connection.createChannel(false);
                channel.basicQos(0, 1, false); // //公平分发、同一时间只处理一个消息。

            }

            // 1 通过basicGet获取原始数据
            // 2 将原始数据转换为特定类型的包
            // 3 处理数据
            // 4 手动发送ack确认 会有三种处理结果：
            //  (4.1)、处理成功，这种时候应该用basicAck确认消息；
            //  (4.2)、可重试的处理失败，这时候应该用basicNack将消息重新入列；
            //  (4.3)、不可重试的处理失败，这时候应该使用basicNack将消息丢弃。
            @Override
            public DetailRes consume() {
                return commConsume();
            }

            private DetailRes commConsume() {
                try {
                    if (!poolExecutorProcess.checkPoolIde()) {
                        log.info("当前队列:{" + queue + "}正在忙碌中...歇会...处理完了再拉取....");
                        try {
                            Thread.sleep(MqConstants.ONE_SECOND);
                        } catch (InterruptedException e) {
                        }
                        // 非空闲 空闲在拉取数据
                        return new DetailRes(true, "当前队列:{" + queue + "}正在忙碌中...");
                    }
                    //1
                    GetResponse response = channel.basicGet(queue, false);

                    while (response == null) {
                        response = channel.basicGet(queue, false);
                        Thread.sleep(MqConstants.ONE_SECOND);
                    }
                    //2

                    T messageBean = JSON.parseObject(new String(response.getBody(), "UTF-8"), clazz);


                    //3
                    DetailRes detailRes;

                    try {
                        poolExecutorProcess.process(messageBean);
                        detailRes = new DetailRes(true, "");
                    } catch (Exception e) {
                        log.error("exception", e);
                        detailRes = new DetailRes(false, "process exception: " + e);
                    }

                    //4
                    if (detailRes.isSuccess()) {
                        channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
                    } else {
                        //避免过多失败log
                        Thread.sleep(MqConstants.ONE_SECOND);
                        log.info("process message failed: " + detailRes.getErrMsg());
                        channel.basicNack(response.getEnvelope().getDeliveryTag(), false, true);

                    }

                    return detailRes;
                } catch (InterruptedException e) {
                    log.error("InterruptedException", e);
                    return new DetailRes(false, "interrupted exception " + e.toString());
                } catch (ShutdownSignalException | ConsumerCancelledException | IOException e) {
                    log.error("exception", e);

                    try {
                        channel.close();
                    } catch (IOException | TimeoutException ex) {
                        log.error("exception", ex);
                    }

                    channel = connection.createChannel(false);

                    return new DetailRes(false, "shutdown or cancelled exception " + e.toString());
                } catch (Exception e) {
                    log.info("exception : ", e);

                    try {
                        channel.close();
                    } catch (IOException | TimeoutException ex) {
                        ex.printStackTrace();
                    }

                    channel = connection.createChannel(false);

                    return new DetailRes(false, "exception " + e.toString());
                }
            }

        };
    }


    private void buildQueue(String exchange, String routingKey,
                            final String queue, Connection connection, ExchangeTypeEnum type, boolean autoDelete) throws IOException {
        Channel channel = connection.createChannel(false);
        if (type.equals(ExchangeTypeEnum.DIRECT)) {
            channel.exchangeDeclare(exchange, "direct", true, false, null);
        } else if (type.equals(ExchangeTypeEnum.TOPIC)) {
            channel.exchangeDeclare(exchange, "topic", true, false, null);
        } else if (type.equals(ExchangeTypeEnum.FANOUT)) {
            channel.exchangeDeclare(exchange, "fanout", true, false, null);
        }

        channel.queueDeclare(queue, true, false, false, null);
        if (!type.equals(ExchangeTypeEnum.FANOUT)) {
            channel.queueBind(queue, exchange, routingKey);
        } else {
            channel.queueBind(queue, exchange, "");

        }
//        channel.addReturnListener(new ReturnListener() {
//            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties basicProperties, byte[] body) throws IOException {
//                String message = new String(body);
//                log.info("Basic.return返回的结果是：{}", message);
//            }
//        });
        try {
            channel.close();
        } catch (TimeoutException e) {
            log.info("close channel time out ", e);
        }
    }

    private void buildTopic(String exchange, Connection connection) throws IOException {
        Channel channel = connection.createChannel(false);
        channel.exchangeDeclare(exchange, "topic", true, false, null);
    }

    private void buildFanout(String exchange, Connection connection) throws IOException {
        Channel channel = connection.createChannel(false);
        channel.exchangeDeclare(exchange, "fanout", true, false, null);
    }

    //for test
    public int getMessageCount(final String queue) throws IOException {
        Connection connection = connectionFactory.createConnection();
        final Channel channel = connection.createChannel(false);
        final AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive(queue);

        return declareOk.getMessageCount();
    }

    public void buildSendQueue(String exchange, String routingKey,
                               final String queue, ExchangeTypeEnum type, boolean autoDelete) {
        RabbitAdmin admin = new RabbitAdmin(connectionFactory);
        if (type.equals(ExchangeTypeEnum.DIRECT)) {
            Queue adminQueue = new Queue(queue);
            admin.declareQueue(adminQueue);
            DirectExchange directExchange = new DirectExchange(exchange, true, autoDelete, null);
            admin.declareExchange(directExchange);
            admin.declareBinding(
                    BindingBuilder.bind(adminQueue).to(directExchange).with(routingKey));
        } else if (type.equals(ExchangeTypeEnum.TOPIC)) {
            TopicExchange topicExchange = new TopicExchange(exchange, true, autoDelete, null);
            admin.declareExchange(topicExchange);
        } else if (type.equals(ExchangeTypeEnum.FANOUT)) {
            FanoutExchange fanoutExchange = new FanoutExchange(exchange, true, autoDelete, null);
            admin.declareExchange(fanoutExchange);
        }
    }

    public void buildConsumerQueue(String exchange, String routingKey,
                                   final String queue, ExchangeTypeEnum type, boolean autoDelete) {
        RabbitAdmin admin = new RabbitAdmin(connectionFactory);
        Queue adminQueue = new Queue(queue);
        admin.declareQueue(adminQueue);

        if (type.equals(ExchangeTypeEnum.DIRECT)) {
            DirectExchange directExchange = new DirectExchange(exchange, true, autoDelete, null);
            admin.declareExchange(directExchange);
            admin.declareBinding(
                    BindingBuilder.bind(adminQueue).to(directExchange).with(routingKey));
        } else if (type.equals(ExchangeTypeEnum.TOPIC)) {
            TopicExchange topicExchange = new TopicExchange(exchange, true, autoDelete, null);
            admin.declareExchange(topicExchange);
            admin.declareBinding(
                    BindingBuilder.bind(adminQueue).to(topicExchange).with(routingKey));

        } else if (type.equals(ExchangeTypeEnum.FANOUT)) {
            FanoutExchange fanoutExchange = new FanoutExchange(exchange, true, autoDelete, null);
            admin.declareExchange(fanoutExchange);
            admin.declareBinding(
                    BindingBuilder.bind(adminQueue).to(fanoutExchange));
        }
    }

}